﻿
using RdKafka;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace BCGPS.DataReceive.Service
{
    public class KafakaProducer
    {
        static string brokerList = string.Empty;
        static string topicName = string.Empty;

        static  KafakaProducer()
        {
            brokerList = ConfigurationManager.AppSettings["Kafka:BrokerList"];
            topicName = ConfigurationManager.AppSettings["Kafka:TopicName"];
            Produce();
        }

        private static void Produce()
        { 
            TopicConfig topicConfig = new TopicConfig() 
            {
                CustomPartitioner = (top, key, cnt) => 
                {
                    var kt = (key != null) ? Encoding.UTF8.GetString(key, 0, key.Length) : "(null)";
                    int partition = 0;
                    if (key != null)
                        partition = key.Length % cnt;
                    bool available = top.PartitionAvailable(partition);
                    Console.WriteLine($"Partitioner topic:{top.Name}   key:{kt}  partition count:{cnt} {partition} {available}");
                    return partition;
                }
            }; 

            Producer producer = new Producer(brokerlist);
            Topic topic = producer.Topic(topicName, topicConfig);

             
        }

        public static SendMsg(string msg)
        { 
            string text;
            while ((text = Console.ReadLine()) != "q")
            {

                byte[] data = Encoding.UTF8.GetBytes(text);


                Task<DeliveryReport> deliveryReport = topic.Produce(data);
                var unused = deliveryReport.ContinueWith(task =>
                {
                    Console.WriteLine($"Partition: {task.Result.Partition} , Offset: {task.Result.Offset}  ");
                });
            }
        }
    }
}
